package com.ww.exercise.coroutine.mid.q15

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.Semaphore
import kotlinx.coroutines.sync.withPermit
import java.lang.management.ManagementFactory
import java.util.concurrent.atomic.AtomicInteger
import java.util.concurrent.atomic.AtomicReference
import kotlin.random.Random
import kotlin.system.measureTimeMillis

// 系统负载指标
data class SystemMetrics(
    val cpuUsage: Double, // CPU使用率
    val memoryUsage: Double // 内存使用率
)

// 日志分析任务结果
data class LogAnalysisResult(
    val taskId: Int,
    val durationMs: Long,
    val processedLines: Int,
    val errors: Int
)

// 监控工具类（单例）
object SystemMonitor {
    private val osBean = ManagementFactory.getOperatingSystemMXBean()
    private val totalMemory = Runtime.getRuntime().totalMemory()

    // 获取当前系统负载指标
    fun getSystemMetrics(): SystemMetrics {
        //计算CPU使用率（最近1秒的平均值）
        val cpuUsage = try {
            val sunOsBean = osBean as? com.sun.management.OperatingSystemMXBean
            sunOsBean?.processCpuLoad?.times(100) ?: 0.0
        } catch (e: Exception) {
            0.0
        }

        // 计算内存使用率
        val usedMemory = totalMemory - Runtime.getRuntime().freeMemory()
        val memoryUsage = (usedMemory.toDouble() / totalMemory) * 100

        return SystemMetrics(cpuUsage, memoryUsage)
    }
}

// 日志分析任务
class LogAnalysisTask(private val taskId: Int) {
    // 模拟日志分析工具
    suspend fun execute(): LogAnalysisResult {
        return withContext(Dispatchers.IO) {
            val startTime = System.currentTimeMillis()

            // 模拟处理100-500ms
            val processingTime = Random.nextLong(100, 500)

            // 模拟处理过程
            delay(processingTime)
            val processedLines: Int = Random.nextInt(1000, 5000)
            val errors: Int = if (Random.nextDouble() < 0.05) Random.nextInt(1, 5) else 0 // 5%概率出现错误

            LogAnalysisResult(
                taskId = taskId,
                durationMs = System.currentTimeMillis() - startTime,
                processedLines = processedLines,
                errors = errors
            )
        }
    }
}

// 动态并发控制器
class DynamicConcurrencyController(
    private val scope: CoroutineScope,
    private val minConcurrency: Int = 5,
    private val maxConcurrency: Int = 50,
    private val initialConcurrency: Int = 20,
    private val adjustmentIntervalMs: Long = 200 // 2秒调整一次
) {
    // 原子引用存储当前信号量，支持线程安全替换
    private val currentSemaphore = AtomicReference(Semaphore(initialConcurrency))

    // 使用原子变量确保并发安全
    private val currentConcurrency = AtomicInteger(initialConcurrency)
    private var isRunning = true

    // 启动监控协程，动态调整并发数
    fun startMonitoring() {
        scope.launch(Dispatchers.Default) {
            while (isRunning) {
                delay(adjustmentIntervalMs)
                adjustConcurrencyBasedOnLoad()
            }
        }
    }

    private fun adjustConcurrencyBasedOnLoad() {
        val metrics = SystemMonitor.getSystemMetrics()
        val current = currentConcurrency.get()
        var newConcurrency = current

        when {
            // 高负载：减少并发（每次减5，不低于最小值）
            metrics.cpuUsage > 80 || metrics.memoryUsage > 70 -> {
                newConcurrency = maxOf(minConcurrency, current - 5)
            }
            // 低负载：增加并发（每次加5，不高于最大值）
            metrics.cpuUsage < 50 || metrics.memoryUsage < 50 -> {
                newConcurrency = minOf(maxConcurrency, current + 5)
            }
        }

        // 并发数变化时更新信号量
        if (newConcurrency != current) {
            currentConcurrency.set(newConcurrency)
            currentSemaphore.set(Semaphore(newConcurrency)) // 替换为新信号量
            println(
                "📊 并发数调整: $current → $newConcurrency (CPU: ${"%.1f".format(metrics.cpuUsage)}%, 内存: ${
                    "%.1f".format(
                        metrics.memoryUsage
                    )
                }%)"
            )
        }
    }

    // 执行受并发控制的任务
    suspend fun <T> execute(block: suspend () -> T): T {
        // 获取当前信号量
        val semaphore = currentSemaphore.get()
        return semaphore.withPermit { // 每次获取1个许可
            block()
        }
    }

    // 停止监控
    fun stop() {
        isRunning = false
    }

    // 获取当前并发数
    fun getCurrentConcurrency() = currentConcurrency.get()
}

// 日志分析管理器
class LogAnalysisManager(private val scope: CoroutineScope) {
    // 运行固定并发数分析
    @OptIn(ExperimentalCoroutinesApi::class)
    suspend fun runFixedConcurrencyAnalysis(
        taskCount: Int,
        concurrency: Int
    ): Pair<Long, List<LogAnalysisResult>> {
        println("===== 启动固定并发分析 (并发数: $concurrency) =====")
        val semaphore = Semaphore(concurrency)
        val results = mutableListOf<Deferred<LogAnalysisResult>>()

        val totalTime = measureTimeMillis {
            // 创建所有任务
            for (i in 0 until taskCount) {
                val task = LogAnalysisTask(i)
                results += scope.async {
                    semaphore.withPermit {
                        task.execute()
                    }
                }
            }
            // 等待所有任务完成
            results.awaitAll()
        }

        return Pair(totalTime, results.map { it.getCompleted() })
    }

    // 运动动态并发分析
    @OptIn(ExperimentalCoroutinesApi::class)
    suspend fun runDynamicConcurrencyAnalysis(
        taskCount: Int
    ): Pair<Long, List<LogAnalysisResult>> {
        println("\n===== 启动动态并发分析 =====")
        val controller = DynamicConcurrencyController(scope)
        val results = mutableListOf<Deferred<LogAnalysisResult>>()

        // 启动监控
        controller.startMonitoring()

        val totalTime = measureTimeMillis {
            // 创建所有任务
            for (i in 0 until taskCount) {
                val task = LogAnalysisTask(i)
                // 显式指定泛型类型
                val deferred = scope.async<LogAnalysisResult> {
                    controller.execute {
                        task.execute()
                    }
                }
                results.add(deferred)
            }
            // 等待所有任务完成
            results.awaitAll()
        }

        controller.stop()
        return Pair(totalTime, results.map { it.getCompleted() })
    }

    // 生成分析报告
    fun generateReport(name: String, durationMs: Long, results: List<LogAnalysisResult>) {
        val totalLines = results.sumOf { it.processedLines }
        val totalErrors = results.sumOf { it.errors }
        val avgDuration = results.map { it.durationMs.toDouble() }.average()
        val minDuration = results.minOf { it.durationMs }
        val maxDuration = results.maxOf { it.durationMs }

        println("\n===== $name 分析报告 =====")
        println("总耗时: ${durationMs}ms (${durationMs / 1000.0}秒)")
        println("处理任务数: ${results.size}")
        println("处理总行数: $totalLines")
        println("错误总数: $totalErrors")
        println("平均任务耗时: ${String.format("%.2f", avgDuration)}ms")
        println("最小任务耗时: $minDuration ms")
        println("最大任务耗时: $maxDuration ms")
        println("==========================")
    }
}

// 主函数
fun main() = runBlocking {
    val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
    val manager = LogAnalysisManager(scope)
    val taskCount = 1000

    // 固定并发分析
    val (fixedTime, fixedResults) = manager.runFixedConcurrencyAnalysis(taskCount, 20)
    manager.generateReport("固定并发模式(20)", fixedTime, fixedResults)

    // 等待系统恢复
    delay(5000)

    // 动态并发分析
    val (dynamicTime, dynamicResults) = manager.runDynamicConcurrencyAnalysis(taskCount)
    manager.generateReport("动态并发模式(5-50)", dynamicTime, dynamicResults)

    // 对比结果
    println("\n===== 模式对比 =====")
    println("固定并发总耗时: ${fixedTime}ms")
    println("动态并发总耗时: ${dynamicTime}ms")
    val timeDiff = dynamicTime - fixedTime
    when {
        timeDiff < 0 -> println("动态模式快于固定模式: ${-timeDiff}ms (${(-timeDiff.toDouble() / fixedTime * 100).toInt()}%)")
        timeDiff > 0 -> println("动态模式慢于固定模式: $timeDiff ms (${(timeDiff.toDouble() / fixedTime * 100).toInt()}%)")
        else -> println("两种模式耗时相同")
    }

    scope.cancel()
    println("\n所有任务完成，系统退出")
}